
#include "RtBase.h"
#include "RtConnectorThreadProxy.h"
#include "RtTransportThreadProxy.h"

//////////////////////////////////////////////////////////////////////
// class CRtConnectorThreadProxy
//////////////////////////////////////////////////////////////////////

CRtConnectorThreadProxy::
CRtConnectorThreadProxy(CRtConnectionManager::CType aType, 
						CRtThread *aThreadNetwork,
						CRtThread *aThreadUser)
	: CRtAcceptorConnectorSinkThreadProxyT<CRtConnectorThreadProxy>(this)
	, m_pThreadUser(aThreadUser)
	, m_pThreadNetwork(aThreadNetwork)
	, m_Type(aType)
{
	if (!m_pThreadNetwork) {
		m_pThreadNetwork = CRtThreadManager::Instance()->GetThread(CRtThreadManager::TT_NETWORK);
		RT_ASSERTE(m_pThreadNetwork);
	}
	if (!m_pThreadUser) {
		m_pThreadUser = CRtThreadManager::Instance()->GetThread(CRtThreadManager::TT_CURRENT);
		RT_ASSERTE(m_pThreadUser);
	}
	RT_ASSERTE(m_pThreadUser != m_pThreadNetwork);
}

CRtConnectorThreadProxy::~CRtConnectorThreadProxy()
{
	// the current thread is network thread due to <m_pConActual>.
//	RT_ASSERTE(CRtThreadManager::IsEqualCurrentThread(m_pThreadNetwork->GetThreadId()));
}

DWORD CRtConnectorThreadProxy::AddReference()
{
	return CRtReferenceControlMutilThread::AddReference();
}

DWORD CRtConnectorThreadProxy::ReleaseReference()
{
	return CRtReferenceControlMutilThread::ReleaseReference();
}

void CRtConnectorThreadProxy::OnReferenceDestory()
{
	// this assert helps to debug that the upper layer 
	// didn't call CancelConnect() before.
	RT_ASSERTE(CRtStopFlag::m_bStoppedFlag);

	// only the user thread can delete this due to <m_pConActual>,
	// so we have to post delete event to the network thread.
	if (CRtThreadManager::IsEqualCurrentThread(m_pThreadUser->GetThreadId())) {
//		RT_INFO_TRACE_THIS("CRtTransportThreadProxy::OnReferenceDestory,"
//			" post delete event to the network thread");
		CRtEventDeleteT<CRtConnectorThreadProxy> *pEventDelete;
		pEventDelete = new CRtEventDeleteT<CRtConnectorThreadProxy>(this);
		if (pEventDelete)
			pEventDelete->Launch(m_pThreadNetwork);
	}
	else {
		delete this;
	}
}

BOOL CRtConnectorThreadProxy::IsConnector()
{
	return TRUE;
}

void CRtConnectorThreadProxy::SetStopFlag()
{
	CRtAcceptorConnectorSinkThreadProxyT<CRtConnectorThreadProxy>::ResetSink(NULL);
	CRtStopFlag::SetStopFlag();
}

void CRtConnectorThreadProxy::
AsycConnect(IRtAcceptorConnectorSink *aSink,  
			const CRtInetAddr &aAddrPeer, 
			CRtTimeValue *aTimeout,
			CRtInetAddr *aAddrLocal)
{
	RT_ASSERTE(CRtThreadManager::IsEqualCurrentThread(m_pThreadUser->GetThreadId()));
	RT_ASSERTE(CRtStopFlag::IsFlagStopped());

	// Don't check m_pConActual due to its operations are all in network thread.
//	RT_ASSERTE(!m_pConActual);

//	RT_INFO_TRACE_THIS("CRtConnectorThreadProxy::AsycConnect,"
//		" aSink=" << aSink <<
//		" addr=" << aAddrPeer.GetIpDisplayName() << 
//		" port=" << aAddrPeer.GetPort() << 
//		" sec=" << (aTimeout ? aTimeout->GetSec() : -1) << 
//		" usec=" << (aTimeout ? aTimeout->GetUsec() : -1));

	RT_ASSERTE(!CRtAcceptorConnectorSinkThreadProxyT<CRtConnectorThreadProxy>::m_pSinkActual);
	RT_ASSERTE(aSink);
	CRtAcceptorConnectorSinkThreadProxyT<CRtConnectorThreadProxy>::ResetSink(aSink);

	RtResult rv = RT_ERROR_OUT_OF_MEMORY;
	CEventAsycConnect *pEvent = new CEventAsycConnect(
		this, this, aAddrPeer, aTimeout, aAddrLocal);
	if (pEvent) {
		rv = m_pThreadNetwork->GetEventQueue()->PostEvent(pEvent);
	}

	if (RT_SUCCEEDED(rv))
		CRtStopFlag::SetStartFlag();

	// TODO: OnConnecteIndication(FAILED) if RT_FAILED(rv)
	RT_ASSERTE(RT_SUCCEEDED(rv));
}

void CRtConnectorThreadProxy::CancelConnect()
{
	RT_ASSERTE(CRtThreadManager::IsEqualCurrentThread(m_pThreadUser->GetThreadId()));

	if (CRtStopFlag::IsFlagStopped())
		return;

	// invoke CRtConnectorThreadProxy::SetStopFlag() in order to ResetSink(NULL).
//	CRtStopFlag::SetStopFlag();
	CRtConnectorThreadProxy::SetStopFlag();

//	RT_INFO_TRACE_THIS("CRtConnectorThreadProxy::CancelConnect,");

	CEventCancelConnect *pEvent = new CEventCancelConnect(this);
	m_pThreadNetwork->GetEventQueue()->PostEvent(pEvent);
}


//////////////////////////////////////////////////////////////////////
// class CEventAsycConnect
//////////////////////////////////////////////////////////////////////

CEventAsycConnect::
CEventAsycConnect(CRtConnectorThreadProxy *aConnectorThreadProxy, 
				  IRtAcceptorConnectorSink *aSink, 
				  const CRtInetAddr &aAddrPeer, 
				  CRtTimeValue *aTimeout,
				  CRtInetAddr *aAddrLocal)
	: m_pOwnerThreadProxy(aConnectorThreadProxy)
	, m_pSink(aSink)
	, m_addrPeer(aAddrPeer)
	, m_pParaTimeout(NULL)
	, m_pParaAddrLocal(NULL)
{
	RT_ASSERTE(m_pOwnerThreadProxy);
	if (aTimeout) {
		m_tvTimeout = *aTimeout;
		m_pParaTimeout = &m_tvTimeout;
	}
	if (aAddrLocal) {
		m_addrLocal = *aAddrLocal;
		m_pParaAddrLocal = &m_addrLocal;
	}
}

CEventAsycConnect::~CEventAsycConnect()
{
}

RtResult CEventAsycConnect::OnEventFire()
{
	RT_ASSERTE(CRtThreadManager::IsEqualCurrentThread(
		m_pOwnerThreadProxy->m_pThreadNetwork->GetThreadId()));

	// we must new actual connector in the network thread.
	// <m_pConActual> may not be NULL due to OnConnectIndicatoin don't assign it to NULL.
//	RT_ASSERTE(!m_pOwnerThreadProxy->m_pConActual);

	CRtConnectionManager::CType type = m_pOwnerThreadProxy->m_Type;

	RtResult rv = CRtConnectionManager::Instance()->CreateConnectionClient(
		type, 
		m_pOwnerThreadProxy->m_pConActual.ParaOut());
	if (RT_FAILED(rv)) {
		RT_ERROR_TRACE_THIS("CEventAsycConnect::OnEventFire,"
			" can't create connector in the network thread. rv=" << rv);
		RT_ASSERTE(FALSE);
		return rv;
	}

	if (m_pOwnerThreadProxy->m_pConActual) {
		m_pOwnerThreadProxy->m_pConActual->
			AsycConnect(m_pSink, m_addrPeer, m_pParaTimeout, m_pParaAddrLocal);
		return RT_OK;
	}
	else
		return RT_ERROR_NULL_POINTER;
}


//////////////////////////////////////////////////////////////////////
// class CEventCancelConnect
//////////////////////////////////////////////////////////////////////

CEventCancelConnect::
CEventCancelConnect(CRtConnectorThreadProxy *aConnectorThreadProxy)
	: m_pOwnerThreadProxy(aConnectorThreadProxy)
{
	RT_ASSERTE(m_pOwnerThreadProxy);
}

CEventCancelConnect::~CEventCancelConnect()
{
}

RtResult CEventCancelConnect::OnEventFire()
{
	RT_ASSERTE(CRtThreadManager::IsEqualCurrentThread(
		m_pOwnerThreadProxy->m_pThreadNetwork->GetThreadId()));

	RT_ASSERTE(m_pOwnerThreadProxy->m_pConActual);
	if (m_pOwnerThreadProxy->m_pConActual) {
		m_pOwnerThreadProxy->m_pConActual->CancelConnect();
		m_pOwnerThreadProxy->m_pConActual = NULL;
		return RT_OK;
	}
	else
		return RT_ERROR_NULL_POINTER;
}

